package com.sunlands.deskmate.channel;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.sunlands.deskmate.client.AuthClient;
import com.sunlands.deskmate.client.LiveVideoService;
import com.sunlands.deskmate.common.annotation.AutoLog;
import com.sunlands.deskmate.constant.YN;
import com.sunlands.deskmate.message.entity.Messages;
import com.sunlands.deskmate.message.handler.receiver.MessageReceiverFactory;
import com.sunlands.deskmate.message.handler.sender.trigger.SendMessageTrigger;
import com.sunlands.deskmate.message.handler.sender.trigger.TriggerEvent;
import com.sunlands.deskmate.message.handler.sender.trigger.TriggerEventEnum;
import com.sunlands.deskmate.utils.DateUtil;
import com.sunlands.deskmate.utils.RedisUtil;
import com.sunlands.deskmate.vo.UserOnlineVO;
import com.sunlands.deskmate.vo.response.BusinessResult;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RMapCache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.context.annotation.Scope;
import org.springframework.http.HttpStatus;
import org.springframework.security.oauth2.common.exceptions.InvalidTokenException;
import org.springframework.stereotype.Component;

import javax.servlet.http.HttpServletRequest;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

import static com.sunlands.deskmate.channel.ChannelManager.CHANNEL_INFO;
import static com.sunlands.deskmate.message.enums.MessageType.*;
import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static io.netty.handler.codec.http.HttpUtil.isKeepAlive;
import static io.netty.handler.codec.http.HttpUtil.setContentLength;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
import static org.springframework.util.StringUtils.isEmpty;


/**
 * @author anjl
 */
@Component
@Slf4j
@Scope("prototype")
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

    private WebSocketServerHandshaker handshaker;

    @Autowired
    private AuthClient authClient;

    @Autowired
    private SendMessageTrigger sendDispatcher;

    @Autowired
    private LiveVideoService tzLiveVideoService;

    public static final String PING_MSG = "ping";
    public static final String PONG_MSG = "pong";


    @Autowired
    private RedisUtil redisUtil;

    @Lazy
    @Autowired
    private MessageReceiverFactory messageReceiverFactory;

    @AutoLog
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof FullHttpRequest) {//建立连接的请求
            handleHttpRequest(ctx, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {//WebSocket
            handleWebsocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    private void handleWebsocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {

        if (frame instanceof CloseWebSocketFrame) {//关闭
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }

        if (frame instanceof PingWebSocketFrame) {//ping消息
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName()));
        }

        String request = ((TextWebSocketFrame) frame).text();
        if (PING_MSG.equalsIgnoreCase(request)) {
            ctx.writeAndFlush(new TextWebSocketFrame(PONG_MSG));
            return;
        }

        Messages messages;
        try {
            messages = JSONObject.parseObject(request, Messages.class);
            //TODO兼容旧版，将ack消息类型强制设置为type=501
            if (Objects.nonNull(messages.getAckId())) {
                messages.setType(ACK.getType());
            }

            messages.setCreateTime(DateUtil.getStringDate());

            log.info("1、接受到消息 msgEntity = {}, channel = {}", messages, ctx.channel().attr(CHANNEL_INFO).get());
        } catch (Exception e) {
            log.error("message format error, message = {}", request, e);
            Messages errorMessage = Messages.builder().type(ERROR_FORMAT.getType())
                    .message(ERROR_FORMAT.getName())
                    .build();
            ack(ctx.channel(), errorMessage);
            return;
        }
        //消息处理
        messageReceiverFactory
                .getReceiver(messages.getType())
                .receive(ctx.channel(), messages);

    }

    public void ack(Channel channel, Messages msgEntity) {
        String message = JSONObject.toJSONString(msgEntity);
        log.debug("ack messages = {}", message);
        ChannelFuture future = channel.writeAndFlush(new TextWebSocketFrame(message));
        future.addListener((ChannelFutureListener) channelFuture -> {
            if (!channelFuture.isSuccess()) {
                channelFuture.channel().writeAndFlush(new TextWebSocketFrame(message));
            }
        });
    }


    public static String getRoomKey(String type, String toId) {
        return type.substring(0, 1) + ":" + toId + ":app";
    }


    public Set<Long> getOnlineUserIdByRoomId(Long roomId, Integer type) {
        String key = type + ":" + roomId;
        Set<Object> onlineUserIds = refreshMembers(key);
        Set<Long> userIdsSet = new HashSet();
        for (Object o : onlineUserIds) {
            userIdsSet.add(Long.valueOf(String.valueOf(o)));
        }
        return userIdsSet;
    }

    private String getHeaders(FullHttpRequest request) {
        HttpHeaders entries = request.headers();
        Set<String> names = entries.names();
        StringBuilder headers = new StringBuilder();
        for (String name : names) {
            String value = entries.getAll(name).toString();
            headers.append(name).append(":").append(value).append(",");
        }
        return headers.toString();
    }


    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
        final String websocket = "websocket";
        final String upgrade = "Upgrade";

        // 如果HTTP解码失败，返回HTTP异常
        if (!request.getDecoderResult().isSuccess() || (!websocket.equalsIgnoreCase(request.headers().get(upgrade)))) {
            sendHttpResponse(ctx, request, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
            return;
        }

        QueryStringDecoder queryStringDecoder = new QueryStringDecoder(request.getUri());
        Map<String, List<String>> params = queryStringDecoder.parameters();

        Channel channel = ctx.channel();

        String token = null;
        Long userId = null;
        if (params.containsKey("token")) {
            token = params.get("token").get(0);
        }

        if (params.containsKey("userId")) {
            userId = Long.parseLong(params.get("userId").get(0));
        }

        if (isEmpty(token) && Objects.isNull(userId)) {
            channel.close();
            log.error("invalid token, token is empty {} userId is null ", channel);
            return;
        }

        WebSocketServerHandshakerFactory handshakerFactory = new WebSocketServerHandshakerFactory(null, null, false);
        handshaker = handshakerFactory.newHandshaker(request);

        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(channel);
        } else {
            handshaker.handshake(channel, request);
        }

        log.debug("handshake token={} channel={}", token, channel);
        ChannelInfo channelInfo = buildChannelInfo(userId, token, channel);

        if (channelInfo == null) {
            Messages msgEntity = new Messages();
            msgEntity.setType(NO_AUTH.getType());
            ctx.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(msgEntity, SerializerFeature.WriteMapNullValue)));
            channel.close();
            log.warn("invalid token {}", token);
            return;
        }
        log.info("用户请求连接通道 channelInfo = {}, header = {}", channelInfo, getHeaders(request));
        ChannelManager.add(channelInfo, channel);

        //不是游客的情况下，推送999拉消息
        if (Objects.nonNull(token)) {
            //每次登入之后发送未读信息
            Messages messages = Messages.builder()
                    .type(NOTICE_EVENT.getType())
                    .realToId(channelInfo.getUserId())
                    .toId(channelInfo.getUserId().toString())
                    .fromUserId(channelInfo.getUserId().toString())
                    .build();
            messageReceiverFactory.getReceiver(NOTICE_EVENT.getType()).receive(channel, messages);
        }

    }

//    private String getIp(ChannelHandlerContext ctx) {
//        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
//        return insocket.getAddress().getHostAddress();
//    }


    private ChannelInfo buildChannelInfo(Long userId, String token, Channel channel) {
        if (Objects.nonNull(token)) {
            String username = null;
            try {
                username = authClient.token(token);
            } catch (InvalidTokenException e) {
                log.info("userId = {} token = {}，but no auth ", userId, token);
            }
            if (Objects.isNull(username)) {
                return null;
            }
            userId = Long.parseLong(username.split("-")[1]);
        }
        log.debug("user info  userId={}, channel={}", userId, channel);
        return new ChannelInfo(userId);
    }

    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
        // 返回应答给客户端
        if (res.getStatus().code() != HttpStatus.OK.value()) {
            ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
            setContentLength(res, res.content().readableBytes());
        }

        // 如果是非Keep-Alive，关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!isKeepAlive(req) || res.getStatus().code() != HttpStatus.OK.value()) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("Exception accoured ", cause);
        try {
            if (ctx != null && ctx.channel() != null && ctx.channel().attr(CHANNEL_INFO) != null &&
                    ctx.channel().attr(CHANNEL_INFO).get() != null) {
                log.info("exceptionCaught accour, userId={}, channelInfo={}", ctx.channel().attr(CHANNEL_INFO).get(), ctx.channel().attr(CHANNEL_INFO).get());
            }
        } catch (Exception e) {
            log.error("exceptionCaught exception", e);
        } finally {
            ctx.close();
        }
    }


    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        ChannelInfo channelInfo = ChannelManager.getChannelInfo(channel);
        if (Objects.nonNull(channelInfo)) {
            TriggerEvent dispatchEvent = new TriggerEvent(TriggerEventEnum.UNSYNC, channelInfo);
            sendDispatcher.unlockEpoch(dispatchEvent);
        }
        log.info("用户通道关闭，触发通道的关闭和从仓库移除 channelInfo = {} ", channelInfo);
        ChannelManager.release(channel);
    }


//    public void inform(MsgChangeInformEntity msgChangeInformEntity) {
//        List<String> userIds = msgChangeInformEntity.getUserIds();
//        for (String userId : userIds) {
//            try {
//                Messages msgEntity = new Messages();
//                msgEntity.setType(msgChangeInformEntity.getType());
//                msgEntity.setToId(userId);
////                sendMessage(userId, "app", msgEntity);
//            } catch (Exception e) {
//                log.error("send inform msg error, userId={}", userId, e);
//            }
//        }
//    }

    public List<UserOnlineVO> getOnline(Long userId) {
        List<UserOnlineVO> userOnlineVOS = new ArrayList<>();

        RMapCache<String, ChannelInfo> redisChannelStore = ChannelManager.getRedisChannelStore();

        if (Objects.isNull(userId)) {
            try {
                return redisChannelStore.readAllValuesAsync().get()
                        .stream()
                        .map(channel ->
                                UserOnlineVO.builder()
                                        .userId(channel.getUserId())
                                        .topRun(channel.getTopRun())
                                        .channelId(channel.getCreateTime())
                                        .backRunTime(channel.getBackRunTime())
                                        .build())
                        .collect(Collectors.toList());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
        ChannelInfo channelInfo = new ChannelInfo(userId);
        ChannelInfo info = redisChannelStore.get(channelInfo.getSessionKey());
        if (Objects.nonNull(info)) {
            UserOnlineVO onlineVO = UserOnlineVO.builder().userId(userId).topRun(info.getTopRun()).backRunTime(info.getBackRunTime()).build();
            userOnlineVOS.add(onlineVO);
        }
        return userOnlineVOS;
    }

    public Map getBusinessOnline(Long userId) {
        Map onlineMapResult = new HashMap();
        Set<Object> keySet = redisUtil.getKeySet(userId.toString());
        for (Object obj : keySet) {
            String key = obj.toString();
            Set<Object> members = refreshMembers(key);
            onlineMapResult.put(key, members);
        }
        return onlineMapResult;
    }

    private Set<Object> refreshMembers(String key) {
        Set<Object> members = redisUtil.members(key);
        log.info("redis get {}", members);
        if ((members == null || members.size() == 0) && key.startsWith("3")) {
            String[] split = key.split(":");
            log.info("房间id是{}", split[1]);
            try {
                BusinessResult<List<Long>> roomUserByRoomId = tzLiveVideoService.getUserIdsByRoomId(0, Long.parseLong(split[1]));
                if (roomUserByRoomId == null || roomUserByRoomId.getData() == null) {
                    return null;
                }
                List<Long> data = roomUserByRoomId.getData();
                data.stream().forEach(o -> redisUtil.addOnlineSet(key, o));
                members = new HashSet(data);
            } catch (Exception e) {
                log.error("获取房间用户失败,{}", e);
            }

        }
        return members;
    }


}
